package com.lmc.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import com.lmc.log.bean.TrafficFlow;
import com.lmc.log.function.FlowReportReduceFunction;
import com.lmc.log.function.FlowReportWindowFunction;
import net.lmc.util.JdbcUtil;
import net.lmc.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

public class JobTrafficTotalFlowJob {
    public static void main(String[] args) throws Exception {

        // 1 - 创建执行环境 - env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2 - 数据源 - source
        DataStream<String> trafficStream = KafkaUtil.consumerKafka(env, "traffic_topic");
        //trafficStream.print("flow");

        // 3 - 数据转换 - transformation
        DataStream<String> resultStream = handle(trafficStream);
        resultStream.print();

        // 4 - 数据接收器 - sink
        JdbcUtil.sinkToClickhouseUpsert(resultStream,
                "INSERT INTO traffic_monitoring.dws_traffic_flow_report_total(\n" +
                        "    window_start_time, window_end_time,\n" +
                        "    camera_id,\n" +
                        "    license_plate_count, \n" +
                        "    ts\n" +
                        ")\n" +
                        "VALUES (?, ?, ?, ?, ?)");

        // 5 - 执行任务 - execute
        env.execute("JobTrafficTotalFlowJob");

    }

    private static DataStream<String> handle(DataStream<String> pageStream) {
        // todo - 将stream中的每条日志数据封装实体类对象Bean

        SingleOutputStreamOperator<TrafficFlow> cameraIdBeanStream = pageStream.map(new RichMapFunction<String, TrafficFlow>() {

            @Override
            public TrafficFlow map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);

                String cameraId = jsonObject.getString("cameraId");
                Long licensePlateCount = 1L;
//                String  licensePlateCount = jsonObject.getString("license_plate_count");

                return new TrafficFlow(
                        null,
                        null,
                        cameraId,
                        licensePlateCount,
                        System.currentTimeMillis()
                );
            }
        });
//        cameraIdBeanStream.print();

        // todo  - 设置 水位线  事件时间字段
        SingleOutputStreamOperator<TrafficFlow> timeStream = cameraIdBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TrafficFlow>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<TrafficFlow>() {
                            @Override
                            public long extractTimestamp(TrafficFlow element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }
                )
        );

        //timeStream.print();

        // todo  - 分组
        KeyedStream<TrafficFlow, String> trafficFlowStringKeyedStream = timeStream.keyBy(value -> value.getCameraId());


        //trafficFlowStringKeyedStream.print();
        // todo  - 开窗 : 滚动窗口,滚动窗口大小为 1 分钟
        WindowedStream<TrafficFlow, String, TimeWindow> windowStream = trafficFlowStringKeyedStream.window(
                TumblingEventTimeWindows.of(Time.minutes(1))
        );


        // todo  - 聚合 : 对窗口中的数据进行聚合
        SingleOutputStreamOperator<String> reportStream = windowStream.reduce(
                // 增量计算                             添加窗口信息
                new FlowReportReduceFunction() , new FlowReportWindowFunction()
        );

        //reportStream.print();


        return reportStream;
    }
}
